package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{KafkaUtil, SparkUtil, DateUtil, NoticeMsg}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 非预警:开启血糖管理计划,用户第一个血糖点表示开启
 * 前3天至少有3次录餐,前10天至少有4天连续血糖数据以及10次录餐,前14天，至少有20次录餐
 *
 */
object BloodSugarFoodDayCountStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))
    val Array(brokers, sugarTopicList, foodTopicList, mealTopicList, mqHost, mqPort, mqUsername,
    mqPassword, mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duation, rddDuration, appName
    ) = args
    //    val checkpointPath = "/data/streaming/checkpoint/bloodsugar_foodday_count_stream/"
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[FoodRecord], classOf[MealRecord]))

    val sc = SparkContext.getOrCreate(conf)

    val updateSugarFunc = (values: Seq[mutable.HashSet[Long]],
                           state: Option[mutable.HashSet[Long]]) => {
      //保存状态
      val currentCount = mutable.HashSet[Long]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(mutable.HashSet[Long]())
      val minTime = if (previousCount.isEmpty) 0L else previousCount.min + 21 * 24 * 60 * 60 * 1000
      currentCount ++= previousCount
      if (currentCount.max > minTime)
        Some(currentCount.filter(_ > minTime))
      else
        Some(currentCount)
    }

    val updateFoodFunc = (values: Seq[mutable.HashMap[Long, Int]],
                          state: Option[mutable.HashMap[Long, Int]]) => {
      //保存状态
      val dateStamp = DateUtil.getTodayDateTime("yyyyMMdd") - 15 * 24 * 60 * 60 * 1000
      val currentCount = mutable.HashMap[Long, Int]()
      values.foreach {
        case map =>
          map.foreach {
            case (day, count) =>
              if (currentCount.contains(day)) currentCount(day) += count
              else currentCount(day) = count
          }
      }
      val previousCount = state.getOrElse(mutable.HashMap[Long, Int]())
      previousCount.foreach {
        case (day, count) =>
          if (day >= dateStamp) {
            if (currentCount.contains(day)) currentCount(day) += count
            else currentCount(day) = count
          }
      }
      Some(currentCount)
    }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duation.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(
        mqHost,
        mqPort.toInt,
        mqUsername,
        mqPassword,
        mqVirtualHost
      )

      val sugarTopics = sugarTopicList.split(",").toSet
      val foodTopics = foodTopicList.split(",").toSet
      val mealTopics = mealTopicList.split(",").toSet
      val sugarMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          sugarTopics
        )
      val foodMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          foodTopics
        )
      val mealMessages = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,mealTopics)

      val sugarBatch: DStream[(String, mutable.HashSet[Long])] = sugarMessages
        .filter(_._2.nonEmpty)
        .flatMap { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val voList = gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              .map { vo =>
                (vo.personId.toString, mutable.HashSet(vo.timestamp))
              }
            voList
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              List(("", mutable.HashSet[Long]()))
          }
        }
        .filter(_._1.nonEmpty)
        .updateStateByKey(updateSugarFunc)

      val foodBatch: DStream[(String, mutable.HashMap[Long, Int])] =
        foodMessages
          .filter(_._2.nonEmpty)
          .map { msg =>
            try {
              println("___________ msg :" + msg)
              val gson = new Gson()
              val vo = gson.fromJson(msg._2, classOf[FoodRecord])
              val dayCount = mutable.HashMap[Long, Int]()
              val recordDate = DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(vo.ateTime))
              dayCount += ((recordDate, 1))
              (vo.personId.toString, dayCount)
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                ("", mutable.HashMap[Long, Int]())
            }
          }
          .filter(_._1.nonEmpty)

      val mealBatch = mealMessages
        .filter(_._2.nonEmpty)
        .map { msg =>
          try {
            println("___________ msg :" + msg._2)
            val gson = new Gson()
            val vo = gson.fromJson(msg._2, classOf[MealRecord])
            val dayCount = mutable.HashMap[Long, Int]()
            val recordDate = DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(vo.recordTime))
            dayCount += ((recordDate, 1))
            (vo.personId.toString, dayCount)
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              ("", mutable.HashMap[Long, Int]())
          }
        }
        .filter(_._1.nonEmpty)
      val foodRdd = foodBatch.union(mealBatch).updateStateByKey(updateFoodFunc)


      sugarBatch.checkpoint(Duration(rddDuration.toInt * 1000))
      foodRdd.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarAndFood =
        sugarBatch.leftOuterJoin(foodRdd).filter(_._2._1.nonEmpty)
      sugarAndFood.foreachRDD { mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition { lines =>
            val todayStamp = DateUtil.getTodayDateTime()
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (sugarList, foodList)) =>
                println("_______ pid:" + pid.toString() + "_______ sugarList:" + sugarList.toString()
                  + "_______ foodList:" + foodList.toString())
                // @todo excel 2: 前3天至少有3次录餐
                if (todayStamp - sugarList.min > 3 * 24 * 60 * 60 * 1000 &&
                  todayStamp - sugarList.min < 3 * 24 * 60 * 60 * 1000 + 12 * 60 * 60 * 1000) {
                  if (foodList.isDefined && foodList.get
                    .filter(_._1 >= sugarList.min)
                    .values
                    .sum >= 3) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.day3notice_a._1,
                      NoticeMsg.day3notice_a._2,
                      NoticeMsg.day3notice_a._3,
                      foodList.get.keys.max.toString,
                      NoticeMsg.day3notice_a._5,
                      NoticeMsg.day3notice_a._4,
                      NoticeMsg.day3notice_a._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                }
                // @todo excel 3: 前14天，至少有20次录餐
                println(s"_______ todayStamp: $todayStamp sugarList.min ${sugarList.min}   todayStamp -   sugarList.min " +
                  s"${todayStamp - sugarList.min}"
                )
                if ((todayStamp - sugarList.min) > 14 * 24 * 60 * 60 * 1000L &&
                  (todayStamp - sugarList.min) < 14 * 24 * 60 * 60 * 1000L + 12 * 60 * 60 * 1000L
                ) {
                  if (foodList.isDefined && foodList.get
                    .filter(_._1 >= sugarList.min)
                    .values
                    .sum >= 20) {
                    println("_____ nimei2:" + true)
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.day14notice_a._1,
                      NoticeMsg.day14notice_a._2,
                      NoticeMsg.day14notice_a._3,
                      foodList.get.keys.max.toString,
                      NoticeMsg.day14notice_a._5,
                      NoticeMsg.day14notice_a._4,
                      NoticeMsg.day14notice_a._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                }
                // @todo excel 1: 前10天至少有4天连续血糖数据以及10次录餐
                if (todayStamp - sugarList.min > 10 * 24 * 60 * 60 * 1000 &&
                  todayStamp - sugarList.min < 10 * 24 * 60 * 60 * 1000 + 12 * 60 * 60 * 1000) {
                  // 连续4天血糖记录
                  var countDay = 1
                  var maxDay = 1
                  val sortDayList = sugarList.toList.sortWith(_.compareTo(_) < 0)
                  var preDay = sortDayList.head
                  for (i <- 1 to sortDayList.length - 1) {
                    val nowDay = sortDayList(i)
                    if (nowDay == preDay + 24 * 60 * 60 * 1000) {
                      countDay += 1
                    } else {
                      maxDay = math.max(maxDay, countDay)
                      countDay = 1
                    }
                    preDay = nowDay
                  }
                  if (maxDay >= 4 && foodList.isDefined && foodList.get
                    .filter(_._1 >= sugarList.min)
                    .values
                    .sum >= 10) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.day10notice_a._1,
                      NoticeMsg.day10notice_a._2,
                      NoticeMsg.day10notice_a._3,
                      DateUtil.getTodayDateTime().toString,
                      NoticeMsg.day10notice_a._5,
                      NoticeMsg.day10notice_a._4,
                      NoticeMsg.day10notice_a._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                }
            }

            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          }
        }
      }

      ssc
    }

    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}